package com.acme.javaschool.intense.sems.sem2concurrency.prodcons;

import java.io.File;
import java.io.FileFilter;
import java.util.ArrayList;
import java.util.concurrent.*;

/**
 * Based on jcip example.
 */
public class ProducerConsumerDequeue {

    private static volatile int filesFound = 0;
    private static volatile int filesIndexed = 0;
    private static final int BOUND = 10;

    public static int getFilesFound() {
        return filesFound;
    }

    public static int getFilesIndexed() {
        return filesIndexed;
    }

    @SuppressWarnings("serial")
    static class DeathPill extends File {
        public DeathPill(String pathname) {
            super("");
        }
    }

    static class FileCrawler implements Runnable {
        private final ArrayList<BlockingDeque<File>> taskRepo;
        private final FileFilter fileFilter;
        private final File root;
        private final CountDownLatch crawlerLatch;
        private int nextDeq;
        private final CountDownLatch startGate;

        public FileCrawler(ArrayList<BlockingDeque<File>> taskRepo,
                           final FileFilter fileFilter,
                           File root,
                           CountDownLatch crawlerLatch,
                           CountDownLatch startGate) {
            this.taskRepo = taskRepo;
            this.root = root;
            this.fileFilter = new FileFilter() {
                public boolean accept(File f) {
                    return f.isDirectory() || fileFilter.accept(f);
                }
            };
            this.crawlerLatch = crawlerLatch;
            this.startGate = startGate;
            nextDeq = 0;
        }

        private boolean alreadyIndexed(File f) {
            return false;
        }

        public void run() {
            try {
                startGate.await();
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }

            try {
                crawl(root);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }

            crawlerLatch.countDown();
        }

        private void crawl(File root) throws InterruptedException {
            File[] entries = root.listFiles(fileFilter);
            if (entries != null) {
                for (File entry : entries)
                    if (entry.isDirectory())
                        crawl(entry);
                    else if (!alreadyIndexed(entry)) {
                        synchronized (taskRepo) {
                            filesFound++;
                        }
                        taskRepo.get(nextDeq).put(entry);
                        nextDeq = (nextDeq + 1) % taskRepo.size();
                    }
            }
        }
    }

    static class Indexer implements Runnable {
        private final ArrayList<BlockingDeque<File>> taskRepo;
        private final int myDeqNum;
        private final CountDownLatch indexerStartGate;
        private int helpDeqNum;
        private Boolean myWorkIsDone;
        private static volatile int pillsFound;
        private static volatile boolean[] workIsDone;
        private static volatile boolean allWorkIsDone;
        private CountDownLatch indexerLatch;

        public Indexer(ArrayList<BlockingDeque<File>> taskRepo, int numberOfDequeue,
                       CountDownLatch indexerStartGate, CountDownLatch indexerLatch) {
            this.taskRepo = taskRepo;
            this.myDeqNum = numberOfDequeue;
            this.indexerStartGate = indexerStartGate;
            this.helpDeqNum = 0;
            this.myWorkIsDone = false;
            this.indexerLatch = indexerLatch;
        }

        public static void initStatic(int numConsumers) {
            pillsFound = 0;
            allWorkIsDone = false;
            workIsDone = new boolean[numConsumers];
            for (int i = 0; i < numConsumers; i++) {
                workIsDone[i] = false;
            }
        }

        public void run() {
            try {
                indexerStartGate.await();
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }

            while (!allWorkIsDone) {
                File file = taskRepo.get(myDeqNum).peek();
                if ((file == null) || (myWorkIsDone)) {
                        do {
                        helpDeqNum = (helpDeqNum + 1) % taskRepo.size();
                    } while ((taskRepo.size() > 1) && (taskRepo.get(helpDeqNum).size() == 0)
                                && (helpDeqNum != myDeqNum) && (workIsDone[helpDeqNum] != false));

                    if (helpDeqNum == myDeqNum) {
                        continue;
                    }
                    work(helpDeqNum);
                } else {
                    work(myDeqNum);
                }
            }
            indexerLatch.countDown();
        }

        public void work(int deqNum) {
            File file = taskRepo.get(deqNum).poll();
            if (file == null) {
                return;
            }

            if (file.getClass() == DeathPill.class) {
                try {
                    taskRepo.get(deqNum).put(file);
                } catch (InterruptedException e) { e.printStackTrace(); }

                synchronized (taskRepo) {
                    if (!allWorkIsDone && !workIsDone[deqNum]) {
                        pillsFound++;
                    }
                    if (deqNum == myDeqNum) {
                        myWorkIsDone = true;
                        workIsDone[myDeqNum] = true;
                    }
                    if (pillsFound >= taskRepo.size()) {
                        allWorkIsDone = true;
                    }
                }
            } else {
                indexFile(file, deqNum);
            }
        }

        public void indexFile(File file, int deqNum) {
            // Index the file...
            if (filesIndexed % 10000 == 0) {
                System.out.print(".");
            }
            synchronized (taskRepo) {
                filesIndexed++;
            }
        };
    }

    public static void startIndexing(File[] roots, final int N_CONSUMERS) {
        filesFound = 0;
        filesIndexed = 0;
        Indexer.initStatic(N_CONSUMERS);

        ArrayList<BlockingDeque<File>> taskRepo = new ArrayList<>();
        for (int i = 0; i < N_CONSUMERS; i++) {
            taskRepo.add(new LinkedBlockingDeque<File>(BOUND));
        }

        FileFilter filter = new FileFilter() {
            public boolean accept(File file) {
                return true;
            }
        };

        CountDownLatch crawlerLatch = new CountDownLatch(roots.length);
        CountDownLatch startGate = new CountDownLatch(1);
        CountDownLatch indexerLatch = new CountDownLatch(N_CONSUMERS);

        for (int i = 0; i < roots.length; i++) {
            new Thread(new FileCrawler(taskRepo, filter, roots[i], crawlerLatch, startGate)).start();
        }

        for (int i = 0; i < N_CONSUMERS; i++) {
            new Thread(new Indexer(taskRepo, i, startGate, indexerLatch)).start();
        }
        startGate.countDown();

        try {
            crawlerLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        for (BlockingDeque<File> deq : taskRepo) {
            try {
                deq.put(new DeathPill(""));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        try {
            indexerLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

}
